package com.seamtop.cuber.server.base;

import com.seamtop.cuber.common.entriy.CuberReducerJob;
import com.seamtop.cuber.common.entriy.CuberReducer;
import com.seamtop.cuber.common.entriy.TransData;
import com.seamtop.cuber.common.util.DateUtil;
import com.seamtop.cuber.core.*;
import io.searchbox.client.JestClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by feng on 2015/8/20.
 */
public class HiveBaseHandler implements BaseAdapter {

    private static final Logger LOG = LoggerFactory.getLogger(HiveBaseHandler.class);

    @Override
    public void execute(CuberReducerJob job)  throws Exception{
        CuberReducer reducer = job.getReducer();
        String hiveSql = HqlAppendUtil.appendHql(reducer,job.getDuration());
        LOG.info("reducer:" + job.getReducer().getReducerId() + " statModel:" + job.getReducer().getStatPeriod() + " duration:" +
                DateUtil.getDateStrOfTimeStamp(job.getDuration().getStartTime()) + "-" + DateUtil.getDateStrOfTimeStamp(job.getDuration().getEndTime()));
        LOG.info("Cuber Hql:"+hiveSql);
        Connection conn = null;
        Statement stmt = null;
        List<TransData> transDataList = new ArrayList<TransData>();
        try {
            conn = HiveManager.getConnetion();
            stmt = conn.createStatement();
            ResultSet res = stmt.executeQuery(hiveSql);
            while (res.next()) {
                TransData transData = HqlAppendUtil.combineResult(res,job);
                transDataList.add(transData);
            }
        }catch (Exception e){
            LOG.error("Reducer[id:"+reducer.getReducerId()+"] execute error -- hive sql execute error",e);
            throw e;
        }finally {
            conn.close();
        }

        if(transDataList != null && transDataList.size()>0){
            for(int i=0;i<transDataList.size();i++){
                TransData transData = transDataList.get(i);
                LOG.info("Put data -- [table:"+reducer.getOutPutTable()+" rowKey:"+transData.getRowKey()+" column:"+transData.getColumnName()+" " +
                        " value:"+transData.getValue()+"]");
            }
            HBaseUtil.addData(transDataList, reducer.getOutPutTable());
            System.out.println("Ok");
            try{
                JestClient client = ESManager.getClient();
                System.out.println("jetclient:"+client);
                ESUtil.index(client, transDataList);
            }catch (Exception e){
                LOG.error("Reducer[id:" + reducer.getReducerId() + "] execute error -- hbase save result error", e);
                throw e;
            }
        }
    }
}
